﻿// --------------------------------------------------------------------------------------------------------------------
// <copyright file="ObservableExtensions.cs" company="XamlNinja">
//   2011 Richard Griffin and Ollie Riches
// </copyright>
// <summary>
//   Observable extensions
// </summary>
// --------------------------------------------------------------------------------------------------------------------

namespace WP7Contrib.Common.Extensions
{
    using System;
    using System.Reactive;
    using System.Reactive.Linq;
    using System.Reactive.Subjects;

    /// <summary>
    /// Observable extensions
    /// </summary>
    public static class ObservableExtensions
    {
        /// <summary>
        /// </summary>
        /// <param name="source">
        /// The source.
        /// </param>
        /// <param name="selector">
        /// The selector.
        /// </param>
        /// <typeparam name="TSource">
        /// </typeparam>
        /// <typeparam name="TOut">
        /// </typeparam>
        /// <returns>
        /// </returns>
        public static IObservable<TOut> Concat<TSource, TOut>(this IObservable<TSource> source, Func<TSource, IObservable<TOut>> selector)
        {
            return Observable.Defer(() =>
            {
                var queue = new BehaviorSubject<Unit>(new Unit());

                return source
                    .Zip(queue, (v, q) => v)
                    .SelectMany(v => selector(v).Do(_ => { }, () => queue.OnNext(new Unit())));
            });
        }

        /// <summary>
        /// </summary>
        /// <param name="source">
        /// The source.
        /// </param>
        /// <param name="minDelay">
        /// The min delay.
        /// </param>
        /// <typeparam name="T">
        /// </typeparam>
        /// <returns>
        /// </returns>
        public static IObservable<T> DelayEach<T>(this IObservable<T> source, TimeSpan minDelay)
        {
            return source.Concat(x =>
              Observable.Empty<T>()
                .Delay(minDelay)
                .StartWith(x));
        }
    }
}
